package com.zlf.sb.demo.consumer;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.zlf.sb.demo.entity.UserJsonDataEntity;
import com.zlf.sb.demo.mapper.UserJsonDataMapper;
import com.zlf.sb.demo.util.SpringUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class InsertDataConsumer implements Runnable {

    private LinkedBlockingQueue<List<UserJsonDataEntity>> linkedBlockingQueue;

    private AtomicInteger count;

    private UserJsonDataMapper mapper;

    private CountDownLatch countDownLatch;

    public InsertDataConsumer(LinkedBlockingQueue<List<UserJsonDataEntity>> linkedBlockingQueue, AtomicInteger count, CountDownLatch countDownLatch) {
        this.linkedBlockingQueue = linkedBlockingQueue;
        this.count = count;
        this.mapper = SpringUtils.getBean(UserJsonDataMapper.class);
        this.countDownLatch = countDownLatch;
    }

    @Override
    public void run() {
        if (CollectionUtil.isNotEmpty(linkedBlockingQueue)) {
            try {
                List<UserJsonDataEntity> dataList = linkedBlockingQueue.take();
                if (CollectionUtil.isNotEmpty(dataList)) {
                    log.info("dataList:{}", JSON.toJSONString(dataList));
                    mapper.insertBatchSomeColumn(dataList);
                    log.info("消费者插入数据完成---->ThreadName:{}", Thread.currentThread().getName());
                    count.getAndAdd(dataList.size());
                }
                countDownLatch.countDown();
            } catch (Exception e) {
                e.printStackTrace();
                log.error("消费者插入数据:{}", e.getMessage());
            }
        }
    }

}
